package com.ibm.cps.processors;

import java.util.Collection;
import java.util.HashSet;
import java.util.Set;

import jline.internal.Nullable;

import com.google.common.base.Function;
import com.google.common.collect.Collections2;
import com.ibm.cps.message.OptimizedParsedDataMessage;
import com.ibm.cps.message.ParsedDataMessage;
import com.ibm.cps.newmessage.AbstractMetadata;
import com.ibm.interfaces.ITimeSeriesSchema;
import com.ibm.util.exception.CPSException;

public abstract class OptimizedProcessor extends AbstractProcessor {

	protected Set<String> subProcessors = new HashSet<String>();

	public OptimizedProcessor(String tenantId, String processorId, String source, ITimeSeriesSchema parentSchema) throws CPSException {
		super(tenantId, processorId, source);
	}

	abstract public void addSubProcessor(AbstractMetadata message)
			throws CPSException;

	abstract public void deleteSubProcessor(String processorid)
			throws CPSException;

	abstract Collection<OptimizedParsedDataMessage> executeSub(
			ParsedDataMessage message) throws CPSException;

	@Override
	public Collection<ParsedDataMessage> execute(ParsedDataMessage message)
			throws CPSException {
		Collection<OptimizedParsedDataMessage> result = executeSub(message);
		if (result == null) {
			return null;
		}
		return Collections2.transform(result,
				new Function<OptimizedParsedDataMessage, ParsedDataMessage>() {
					public ParsedDataMessage apply(
							@Nullable OptimizedParsedDataMessage input) {
						return (ParsedDataMessage) input;
					}
				});
	}

	public Set<String> getSubprocessors() {
		return subProcessors;
	}
}
